package thread.n8;

import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 自定义线程池的测试类
 */
@Slf4j(topic = "c.TestPool")
public class TestPool {
    public static void main(String[] args) {

        MyRejectPolicy<Runnable> objectMyRejectPolicy = new MyRejectPolicy<>();
        MyThreadPool threadPool = new MyThreadPool(2,
                3000,
                TimeUnit.MILLISECONDS,
                10,
                objectMyRejectPolicy
                );
        int total =4;
        for (int i = 1; i <= total; i++) {
            int j = i;
            threadPool.execute(() -> {
                try {
                    System.out.println("开始执行第 {}/{} 个任务 "+j+"："+total);
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("第 {}/{} 个任务 执行结束"+j+"："+total);
            });
        }
    }
}

/**
 * 阻塞队列：专门用于线程池和任务数量的交互的。
 *      比如：现在任务很多，线程池没有空闲线程了，那么这个任务一定会存放在阻塞队列中。
 * @param <T>
 */
class MyBlockingQueue<T>{
    //1.任务队列。这个就是阻塞队列专门存储任务的地方。
    private Deque<T> queue = new ArrayDeque<>();

    //2.锁：这个是防止多个线程同时获取任务时，出现了问题，所以我们需要一个锁，来保护这个任务队列。
    private ReentrantLock lock = new ReentrantLock();

    //3.生产者条件变量：这个是防止生产者向一个满队列中继续添加元素。
    private Condition fullWaitSet = lock.newCondition();

    //4.消费者条件变量：同上，防止消费者在空队列中继续获取元素。
    private Condition emptyWaitSet = lock.newCondition();

    //5：容器容量
    private int capcity;

    public MyBlockingQueue(int capcity) {
        this.capcity = capcity;
    }

    //带时间的阻塞获取
    public T poll(long timeout, TimeUnit unit){
        //将超时时间，统一转换为nanos。
        long l = unit.toNanos(timeout);
        try {
            lock.lock();
            //判断当前队列是否为空
            /**
             * TODO：这个【while循环】是重点，它涉及到两个知识点：
             *      1：虚假唤醒。一句话【条件判断，一定要使用while】，而不是【if】，使用【if】可能出现异常（就是别的线程意外的唤醒了这个阻塞线程，但是条件却不满足）。
             *      2：需要使用【await方法】返回的值，进行重新阻塞（出现虚假唤醒时）。要不然它每次都是使用【最开始的值】进行阻塞，就会出现问题。
             */
            while (queue.isEmpty()){
                try {
                    //如果剩余的时间小于0，则直接返回null。
                    if(l <= 0){
                        return null;
                    }
                    //返回值是剩余的时间。
                    l = emptyWaitSet.awaitNanos(l);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //如果不为空，则返回第一个元素。
            T t = queue.removeFirst();
            //唤醒添加的阻塞线程。
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take(){
        try {
            lock.lock();
            //判断当前队列是否为空
            while (queue.isEmpty()){
                try {
                    //如果为空，则阻塞获取元素的线程。
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //如果不为空，则返回第一个元素。
            T t = queue.removeFirst();
            //唤醒添加的阻塞线程。
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }


    /**
     * 向队列尾部添加一个元素, 阻塞至 加入到队列 或 超时时长
     * @param task
     * @param timeout
     * @param timeUnit
     * @return
     */
    public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            //同poll方法一样的，解决虚假唤醒的代码。
            while (queue.size() == capcity) {
                try {
                    if(nanos <= 0) {
                        return false;
                    }
                    System.out.println("等待加入任务队列："+ task);
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            System.out.println("加入任务队列："+task);
            queue.addLast(task);
            //添加任务成功后,唤起 等待在emptyWaitSet上的 一个线程
            emptyWaitSet.signal();

            return true;
        } finally {
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T t){
        lock.lock();
        try {
            //判断当前队列是否满了。
            while(queue.size() == capcity){
                try {
                    //队列已经满了，阻塞添加的线程。
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //否则添加进最后一位。
            queue.addLast(t);
            //唤醒获取的阻塞线程。
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    //获取大小
    public int size(){
        try {
            lock.lock();
            return queue.size();
        }finally {
            lock.unlock();
        }
    }

    /**
     * 代失败策略的存放方法，因为我们的任务队列不是无限大的。一定会有满的时候。
     * 尝试将任务task放入队列, 如果队列已满,则执行拒绝策略rejectPolicy
     */
    public void tryPut(MyRejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            // 判断队列是否满
            if(queue.size() == capcity) {
                System.out.println("队列已满,按照拒绝策略处理任务："+task);
                rejectPolicy.reject(this, task);
            } else {  // 有空闲
                System.out.println("队列未满,任务 "+task+" 加入到队列中。");
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }

}

/**
 * 拒绝策略：这里实现了一个最简单的策略，满了就遗弃策略。
 * @param <T>
 */
class MyRejectPolicy<T>{
    public void reject(MyBlockingQueue t1,T t){
        System.out.println("当前任务队列："+t1+"，已满，拒绝接收新的任务。。。"+t);
    }

    //一些其他的拒绝策略。
//    (queue, task)->{
//        // 1. 死等
//        queue.put(task);
//        // 2) 带超时等待
////                    queue.offer(task, 1500, TimeUnit.MILLISECONDS);
//        // 3) 让调用者放弃任务执行
//        System.out.println("放弃任务："+task);
//        // 4) 让调用者抛出异常
////            throw new RuntimeException("任务执行失败 " + task);
//        // 5) 让调用者自己执行任务
//        System.out.println("当前拒绝策略: 让调用线程池的调用者自己执行任务,没有开新线程,直接调用的run()");
////                task.run();
//    }

}

/**
 * 自定义的线程池：里面应该具有如下属性，这只是一个简单的线程池，实际上java中的线程池还有一些额外的参数。
 *      1：线程池核心数。
 *      2：阻塞队列。
 *      3：进入阻塞队列时的失败拒接策略。
 *      4：线程集合。
 *      5：获取任务时的超时时间等。
 */
class MyThreadPool {

    // 任务队列
    private MyBlockingQueue<Runnable> taskQueue;
    //队列已满时的拒绝策略
    private MyRejectPolicy<Runnable> rejectPolicy;

    // 线程集合
    private HashSet<Worker> workers = new HashSet<>();

    // 核心线程数
    private int coreSize;

    // 获取任务时的超时时间
    private long timeout;
    private TimeUnit timeUnit;


    // 执行任务
    public void execute(Runnable task) {
        System.out.println("接收到任务需要执行: "+task);

        // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行
        // 如果任务数超过 coreSize 时，加入任务队列暂存
        synchronized (workers) {
            if(workers.size() < coreSize) {
                Worker worker = new Worker(task,"worker--"+workers.size());
                workers.add(worker);
                System.out.println("coreSize未满,新增 worker："+worker+"，来执行任务："+task);
                worker.start();

            } else {
                System.out.println("coreSize已经满了!!!!!,需要先尝试将任务："+task+"，放到等待队列中 ");
                taskQueue.tryPut(rejectPolicy, task);

            }
        }
    }

    /**
     * 构造函数
     *
     * @param coreSize 线程池最大核心线程数
     * @param timeout 和timeUnit一起指定超时时长
     * @param timeUnit 和timeout一起指定超时时长
     * @param queueCapcity 任务队列容量
     * @param rejectPolicy 任务队列满时针对添加操作的拒绝策略
     */
    public MyThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, MyRejectPolicy<Runnable> rejectPolicy) {
        System.out.println("构造ThreadPool");
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new MyBlockingQueue<>(queueCapcity);
        this.rejectPolicy = rejectPolicy;
    }

    /**
     * 线程池中的工作线程
     */
    class Worker extends Thread{
        /**
         * 执行任务主体
         */
        private Runnable task;

        public Worker(Runnable task,String workerName) {
            this.task = task;
            this.setName(workerName);
        }

        /**
         * 执行已有任务或从队列中获取一个任务执行.
         * 如果都执行完了,就结束线程
         */
        @Override
        public void run() {
            System.out.println("worker跑run了,让我看看有没有task来做");

            // 执行任务
            // 1) 当 task 不为空，执行任务
            // 2) 当 task 执行完毕，再接着从任务队列获取任务并执行
//            while(task != null || (task = taskQueue.take()) != null) {
            while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                //take 方法，不会停止线程， poll方法会停止，区别在于，一个是超时等待，另一个不是超时等待。
                try {
                    System.out.println("获取到任务了,正在执行..."+task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("搞定一个任务 {},尝试获取新任务执行"+task);
                    task = null;
                }
            }

            //如果当前线程的任务为空，并且任务队列中没有任务了，则移除这个线程对象。
            synchronized (workers) {
                System.out.println("当前worker："+this+"，因长时间没有获取到可执行任务 将被释放");
                workers.remove(this);
            }

        }

    }

}